package org.budo.canal.instance.spring;

import java.net.InetSocketAddress;
import java.util.List;

import org.budo.canal.message.handler.CanalMessageHandler;
import org.budo.canal.message.handler.DefaultCanalMessageHandler;
import org.budo.canal.parse.inbound.mysql.MysqlEventParserFactoryBean;
import org.budo.support.javax.sql.util.JdbcUtil;
import org.budo.support.lang.util.ReflectUtil;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;

import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.common.alarm.LogAlarmHandler;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.meta.MemoryMetaManager;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
import com.alibaba.otter.canal.store.model.BatchMode;
import com.alibaba.otter.canal.store.model.Event;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @see com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring
 */
@Getter
@Setter
@Slf4j
public abstract class AbstractCanalInstanceFactoryBean implements FactoryBean<CanalInstance>, BeanNameAware {
    private BatchMode batchMode = BatchMode.MEMSIZE;

    private int bufferSize = 2 * 1024;

    private String destination;

    private String beanName;

    /**
     * TODO 这些属性对象，应该不可以让多个 CanalInstance 用同一个
     */
    private CanalMetaManager metaManager = new MemoryMetaManager();

    private CanalAlarmHandler alarmHandler = new LogAlarmHandler();

    private CanalLogPositionManager logPositionManager = new MemoryLogPositionManager();

    private CanalEventStore<Event> eventStore;

    private CanalEventParser eventParser = null;

    private CanalEventSink<List<Entry>> eventSink;

    // 连接相关配置
    private AuthenticationInfo masterInfo;

    private String host;

    private Integer port = 3306;

    private String username;

    private String password = "";

    // 拉取相关配置
    private CanalMessageHandler canalMessageHandler = new DefaultCanalMessageHandler();

    private Long slaveId;

    /**
     * 单次拉消息条数
     * 
     * @see https://github.com/alibaba/otter/blob/master/node/etl/src/main/java/com/alibaba/otter/node/etl/select/selector/canal/CanalEmbedSelector.java
     */
    private Integer batchSize = 1000;

    /**
     * 单次拉消息超时 a. 如果timeout为null，则采用tryGet方式，即时获取 b. 如果timeout不为null 1.
     * timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回 2.
     * timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
     */
    private Long timeout = 1000L;

    public CanalEventStore<Event> getEventStore() {
        if (this.eventStore == null) {
            this.eventStore = this.defaultMemoryEventStoreWithBuffer();
        }
        return this.eventStore;
    }

    /**
     * 作为方法，执行时间在 destination 有值之后
     */
    public CanalEventParser getEventParser() {
        if (null != this.eventParser) {
            return this.eventParser;
        }

        MysqlEventParserFactoryBean eventParser = new MysqlEventParserFactoryBean();
        eventParser.setDestination(this.destination());
        eventParser.setSlaveId(this.getSlaveId());

        if (null != this.getLogPositionManager()) {
            eventParser.setLogPositionManager(this.getLogPositionManager());
        }

        this.setEventParser(eventParser.build());
        return this.eventParser;
    }

    public void setEventParser(CanalEventParser eventParser) {
        if (eventParser instanceof AbstractEventParser //
                && null == ReflectUtil.getFieldValue(AbstractEventParser.class, "destination", eventParser)) {
        }

        this.eventParser = eventParser;
    }

    protected String destination() {
        if (null != this.getDestination()) {
            return this.getDestination();
        }

        return this.getBeanName();
    }

    protected AuthenticationInfo masterInfo() {
        if (null != this.getMasterInfo() //
                && this.checkConnectMysql(this.getMasterInfo())) {
            return this.getMasterInfo();
        }

        if (null != this.getHost() && null != this.getUsername()) {
            InetSocketAddress address = new InetSocketAddress(this.getHost(), this.getPort());
            AuthenticationInfo authenticationInfo = new AuthenticationInfo(address, this.getUsername(), this.getPassword());

            Boolean checkConnectMysql = this.checkConnectMysql(authenticationInfo);
            if (checkConnectMysql) {
                return authenticationInfo;
            }

            log.error("#148 masterInfo, this=" + this);
            return null;
        }

        log.error("#152 masterInfo, this=" + this);
        return null;
    }

    /**
     * 配置检查
     */
    private Boolean checkConnectMysql(AuthenticationInfo authenticationInfo) {
        try {
            JdbcUtil.checkConnectMysql(authenticationInfo.getAddress().getHostName(), //
                    authenticationInfo.getAddress().getPort(), //
                    authenticationInfo.getUsername(), //
                    authenticationInfo.getPassword());

            return true;
        } catch (Throwable e) {
            log.error("#161 authenticationInfo=" + authenticationInfo + ", e=" + e, e);
            return false;
        }
    }

    private MemoryEventStoreWithBuffer defaultMemoryEventStoreWithBuffer() {
        MemoryEventStoreWithBuffer memoryEventStoreWithBuffer = new MemoryEventStoreWithBuffer();
        memoryEventStoreWithBuffer.setRaw(false); // 注意，需要 false
        memoryEventStoreWithBuffer.setDdlIsolation(true);

        memoryEventStoreWithBuffer.setBatchMode(this.getBatchMode()); // 限制内存消耗
        memoryEventStoreWithBuffer.setBufferSize(this.getBufferSize());

        return memoryEventStoreWithBuffer;
    }

    @Override
    public Class<?> getObjectType() {
        return CanalInstance.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }
}
